from pymongo import MongoClient

# 配置信息
MONGODB_DATABASES = {
    "default": {
        "type": "mongodb",
        'user': 'medpeer',
        'password': 'medpeer',
        'auth_source': 'admin',  # 认证数据库
        'host': '192.168.1.245',
        'port': 27017,
        'database': 'raw_data',
        "charset": "utf8mb4"
    },
}


# 连接 MongoDB
def get_mongo_collection():
    """
    建立 MongoDB 连接并返回 raw_cn_yiigle_article_list 集合对象

    :return: MongoDB collection 对象
    """
    cfg = MONGODB_DATABASES["default"]

    # 构建连接 URI
    uri = f"mongodb://{cfg['user']}:{cfg['password']}@{cfg['host']}:{cfg['port']}/{cfg['database']}?authSource={cfg['auth_source']}"

    # 创建客户端
    client = MongoClient(uri)

    # 获取数据库
    db = client[cfg['database']]

    # 返回集合
    return db["raw_information_list_temp"]


from kafka import KafkaProducer
import json
import time

# 1. 把 3 个 broker 地址统一成 kafka-python 需要的格式
bootstrap_servers = [
    '180.76.250.147:19092',
    '180.76.250.147:19096',
    '180.76.250.147:19100'
]

# 2. 创建 Producer（json 序列化）
producer = KafkaProducer(
    bootstrap_servers=bootstrap_servers,
    value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode('utf-8'),
    acks='all',  # 高可靠
    retries=3,  # 自动重试
    max_block_ms=30_000  # 阻塞超时
)

topic = 'temp4'

if __name__ == "__main__":
    # 获取集合对象
    collection = get_mongo_collection()

    # 投影：排除 create_time 等对象字段字段
    projection = {"create_time": 0, 'original_html': 0, "info_html": 0}

    # 执行查询
    doc = collection.find({}, projection)

    for index, item in enumerate(doc):
        item['_id'] = str(item['_id'])
        future = producer.send(topic, item)
        print(f'send => {index}')
        time.sleep(0.2)
        # input("Press Enter to continue...")
    # 5. 等待缓冲区全部发完
    producer.flush()
    producer.close()
    print('done.')